package com.dddframework.mq.impl;

import com.dddframework.core.contract.MQEvent;
import com.dddframework.core.contract.exception.ServiceException;
import com.dddframework.core.utils.JsonKit;
import com.dddframework.mq.core.MQClient;
import com.dddframework.mq.core.MQFilter;
import com.dddframework.mq.core.MQListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * Redis Pubsub客户端实现
 */
@Slf4j(topic = "### BASE-MQ : redisClient ###")
@Component
public final class RedisClient implements MQClient {
    private static Jedis PUBLISHER;
    BlockingQueue<MQEvent> SENDING_MSGS = new LinkedBlockingQueue<>();
    private ExecutorService es = Executors.newSingleThreadExecutor();
    private AtomicBoolean started = new AtomicBoolean(false);

    @Override
    public String impl() {
        return "redis";
    }

    private void run() {
        es.submit(() -> {
            log.info("Started MQ publisher");
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    MQEvent mqEvent = SENDING_MSGS.take();
                    String message = serialization().serialize(mqEvent);
                    // channel=namespace:topic:tag或namespace:topic或topic
                    String channel = String.format("%s%s%s", config().namespace(":"), mqEvent.getTopic(),
                            (mqEvent.getTag() == null || mqEvent.getTag().isEmpty()) ? "" : (":" + mqEvent.getTag()));
                    try {
                        if (PUBLISHER == null) {
                            PUBLISHER = jedis();
                        }
                        PUBLISHER.publish(channel, message);
                        log.info("Publish MQ [{}]: {}", channel, message);
                    } catch (Exception e) {
                        log.error("Publish MQ [{}]: {} failed!", channel, message, e);
                    }
                } catch (InterruptedException e) {
                    log.error("Publish MQ failed!", e);
                }
            }
            log.info("MQ publisher stopped");
        });
    }

    @Override
    public Consumer<MQEvent> initProducer() {
        if (started.compareAndSet(false, true)) {
            run();
        }
        return mqEvent -> SENDING_MSGS.offer(mqEvent);

    }

    @Override
    public boolean initConsumer(MQListener mqListener) throws Exception {
        // channel=namespace:topic:tag或namespace:topic或topic，TODO 暂不支持通配符*和非-
        List<String> channels = new ArrayList<>();
        if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
            Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
            if (!tags.isEmpty()) {
                for (String tag : tags) {
                    channels.add(config().namespace(":") + mqListener.getTopic() + ":" + tag);
                }
            }
        } else {
            channels.add(config().namespace(":") + mqListener.getTopic());
        }
        new Thread(() -> jedis().subscribe(new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                MQEvent mqEvent = mqListener.getDeserialize().apply(message);
                String group = String.format("ChannelGroupLock:%s:%s", mqListener.namespaceTopicTags(), mqEvent.getMsgId());
                try {
                    // 按组消费：Redis本身没有消费组的概念，这里我们锁住某条消息的消费资格10秒钟，以实现同消费组（如多实例）只能由一个消费者消费
//                    if (!singleton().exists(lockKey)) {
//                        singleton().setex(lockKey, 10L, lockKey);
                    // 本条消息在该消费者组未消费
                    consume(mqListener, mqEvent);
//                    }
                } catch (Throwable e) {
                    if (e instanceof ServiceException) {
                        log.error("Consume MQ [{}] failed: {} => {}", group, e.getMessage(), JsonKit.toJson(mqEvent));
                    } else {
                        log.error("Consume MQ [{}] failed: {}", group, mqEvent, e);
                    }
                }
            }

            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                log.info("Subscribed channel: {}", channel);
            }

            @Override
            public void onUnsubscribe(String channel, int subscribedChannels) {
                log.info("Unsubscribed channel: {}", channel);
            }
        }, channels.toArray(new String[0]))).start();
        return true;
    }

    private synchronized Jedis jedis() {
        String[] hostAndPort = config().getServer().split(":");
        if (!config().getPassword().isEmpty()) {
            Jedis jedis = new Jedis(hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000);
            jedis.auth(config().getPassword());
            return jedis;
        } else {
            return new Jedis(hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000);
        }
    }

}